package com.jxsr.kafka.demos.web;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Description : kafka消费者
 * @Author : Bruce Lee
 * @CreateTime : 2024/1/11
 */


@Slf4j
@Component
public class Consumer {

    /**
     * 开启批量消费
     * @param records
     * @param ack
     */
    @KafkaListener(topics = {"test"})
    public void func(List<ConsumerRecord<String, Object>> records, Acknowledgment ack){
        for (ConsumerRecord<String, Object> record : records) {
            String topic = record.topic();
            log.info("订阅topic：{}, 数据内容：{}", topic, record.value());
            // 确认消息已消费
            ack.acknowledge();
        }
    }


    /**
     *
     * 未开启批量消费
     * @param record
     * @param ack
     */
//    @KafkaListener(topics = {"test"})
//    public void func(ConsumerRecord<String, Object> record, Acknowledgment ack){
//        String topic = record.topic();
//        log.info("订阅topic：{}, 数据内容：{}", topic, record.value());
//        // 确认消息已消费
//        ack.acknowledge();
//    }

}
